/*
 * Copyright 2020-8-27 Baheal
 * @(#)RocketMqMessageListenerImpl.java
 */
package com.byyy.ccts.core.message.rocketmq.service.impl;

import com.alibaba.fastjson.JSON;
import com.byyy.ccts.core.message.rocketmq.config.Configuration;
import com.byyy.ccts.core.message.rocketmq.service.RocketMqMessageListener;
import com.byyy.ccts.core.utils.byyyoa.OAModel;
import com.byyy.ccts.core.utils.cq.CQMailUtils;
import com.byyy.ccts.modules.cc.service.IPsZcdService;
import com.byyy.ccts.modules.sys.Constants;
import com.byyy.ccts.modules.sys.entity.TZParam;
import com.byyy.ccts.modules.sys.service.IParamService;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.List;


/**
 * @author liyongbo
 * @version 1.0
 * @date 2020-8-27 15:51
 * @description： 消息处理
 */
@Service
public class RocketMqMessageListenerImpl implements RocketMqMessageListener {

    @Autowired
    private IParamService paramService;

    private static final Logger logger = LoggerFactory.getLogger(RocketMqMessageListenerImpl.class);

    @Autowired
    private IPsZcdService psZcdService;

    @Override
    public boolean onMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        TZParam param = paramService.getParamByCode("SMS");
        for (MessageExt msgExt : messages) {
            String msg = StringUtils.toEncodedString(msgExt.getBody(), Charset.defaultCharset());
            if (Configuration.TAG_BACK_WMS_ZCD.equals(msgExt.getProperty("TAGS"))) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    logger.error("线程休眠发生异常！", e);
                }
                psZcdService.back(Arrays.asList(msg));
            }
            else {
                OAModel model = JSON.parseObject(msg, OAModel.class);
                CQMailUtils.sendMail(model.getCode(), model.getTitle(), model.getContent());
                if (Constants.Y.equals(param.getValue()) && StringUtils.isNotEmpty(model.getExt())) {
                    CQMailUtils.sendMobileMsg(model.getCode(), model.getExt());
                }
            }
            logger.debug("消息处理完毕[{}-{}-{}]", msgExt.getTopic(), msgExt.getTags(), msg);
        }
        return true;
    }
}
